package com.sunchp;

import com.sunchp.utils.SparkUtils;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.api.java.function.VoidFunction;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import scala.Tuple2;

import java.util.Arrays;

public class WordCountCalcLearing {
    private static final Logger logger = LoggerFactory.getLogger(WordCountCalcLearing.class);

    public static void main(String[] args) {
        JavaSparkContext jsc = SparkUtils.getJavaSparkContext("WordCountSpark", "local[2]", "WARN");
        JavaRDD<String> wordRdd = SparkUtils.createRddExternal(jsc, "/Users/sunchp/gitRepo/spark-sample/src/main/resources/README.md");
        wordCountCal(wordRdd);

        jsc.stop();
    }

    public static void wordCountCal(JavaRDD<String> wordRdd) {
        JavaRDD<String> wordFlatMap = wordRdd.flatMap(new FlatMapFunction<String, String>() {
            public Iterable<String> call(String s) throws Exception {
                return Arrays.asList(s.split("[^a-zA-Z']+"));
            }
        });

        // 将每个单词映射次数为1
        JavaPairRDD<String, Integer> wordMapToPair =
                wordFlatMap.mapToPair(new PairFunction<String, String, Integer>() {
                    public Tuple2<String, Integer> call(String s) throws Exception {
                        return new Tuple2<String, Integer>(s, 1);
                    }
                });

        //将每个重复key的value相加
        JavaPairRDD<String, Integer> wordReduceByKey = wordMapToPair.reduceByKey(new Function2<Integer, Integer, Integer>() {
            public Integer call(Integer v1, Integer v2) throws Exception {
                return v1 + v2;
            }
        });

        wordReduceByKey.sortByKey().foreach(new VoidFunction<Tuple2<String, Integer>>() {
            public void call(Tuple2<String, Integer> t) throws Exception {
                logger.warn("key:" + t._1 + ",value:" + t._2);
            }
        });
    }
}
